package com.zyh.study.myForestDemoPlus;


import com.dtflys.forest.annotation.SSEName;
import com.dtflys.forest.annotation.SSEValue;
import com.dtflys.forest.http.ForestSSE;
import com.dtflys.forest.interceptor.SSEInterceptor;
import com.dtflys.forest.sse.EventSource;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.f4b6a3.ulid.Ulid;
import com.github.f4b6a3.ulid.UlidCreator;
import com.zyh.study.myForestDemoPlus.EventCompensationService;
import com.zyh.study.myForestDemoPlus.OrderEventDTO;
import com.zyh.study.myForestDemoPlus.OrderSSEDeal;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class AlpacaSSEDeal implements CommandLineRunner, SSEInterceptor {

    protected static volatile ForestSSE currentSSE;
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private EventCompensationService compensationService;

    @Autowired
    private OrderSSEDeal orderSSEDeal;

    @Autowired
    @Lazy
    private AlpacaSSEDeal alpacaSSEDeal;

    // Redis键定义
    private static final String EVENT_KEY_PREFIX = "alpaca:event:";
    private static final String LAST_PROCESSED_ID_KEY = "alpaca:last_processed_id";
    private static final String CONNECTION_STATUS_KEY = "alpaca:sse:status";

    // 线程池处理事件
    private final ExecutorService eventProcessor = Executors.newFixedThreadPool(4);

    @Override
    public void run(String... args) {
        initializeLastProcessedId();
        connectSSE();

        // 启动时检查未处理事件
        compensationService.checkUnprocessedEvents();
    }

    /**
     * 初始化最后处理的事件ID
     */
    private void initializeLastProcessedId() {
        RBucket<String> lastIdBucket = redissonClient.getBucket(LAST_PROCESSED_ID_KEY);
        if (!lastIdBucket.isExists()) {
            // 初始化为最小ULID: 00000000000000000000000000
            lastIdBucket.set("00000000000000000000000000");
        }
    }

    /**
     * 连接SSE
     */
    public synchronized void connectSSE() {
        if (currentSSE != null) {
            return;
        }
        try {
            RBucket<String> lastIdBucket = redissonClient.getBucket(LAST_PROCESSED_ID_KEY);
            String lastProcessedId = lastIdBucket.get();

            orderSSEDeal.subscribeToOrderStatus(
                    null,
                    null,
                    lastProcessedId,
                    "",
                    alpacaSSEDeal
            );

            // 更新连接状态
            redissonClient.getBucket(CONNECTION_STATUS_KEY).set("connected");
        } catch (Exception e) {
            log.error("SSE connection error: {}", e.getMessage(), e);
            redissonClient.getBucket(CONNECTION_STATUS_KEY).set("disconnected");
            scheduleReconnect();
        }
    }

    /**
     * 处理SSE消息
     */
    private void handleSSEMessage(OrderEventDTO event) {
        if (event == null || event.getEventId() == null) {
            return;
        }
        eventProcessor.submit(() -> {
            try {
                Ulid currentUlid = Ulid.from(event.getEventId());
                Ulid lastUlid = getLastProcessedUlid();
                int comparison = currentUlid.compareTo(lastUlid);

                // 事件处理主逻辑
                if (comparison <= 0) {
                    handleHistoricalEvent(event, currentUlid);
                } else if (comparison == 1) {
                    handleNewEvent(event, currentUlid, lastUlid);
                }
            } catch (Exception e) {
                log.error("Event processing failed: {}", event.getEventId(), e);
            }
        });
    }

    /**
     * 处理历史事件（ID小于等于最后处理ID）
     */
    private void handleHistoricalEvent(OrderEventDTO event, Ulid eventUlid) {
        String eventKey = EVENT_KEY_PREFIX + event.getEventId();
        RBucket<String> eventBucket = redissonClient.getBucket(eventKey);

        // 事件已存在：幂等跳过
        if (eventBucket.isExists()) {
            log.debug("Skipping processed event: {}", event.getEventId());
            return;
        }
        try {
            // 保存并处理历史事件
            eventBucket.set(objectMapper.writeValueAsString(event), 7, TimeUnit.DAYS);
            processEventContent(event);
            log.warn("Processed late historical event: {}", event.getEventId());
        } catch (JsonProcessingException e) {
            log.error("Failed to save historical event: {}", event.getEventId(), e);
        }
    }

    /**
     * 处理新事件（ID大于最后处理ID）
     */
    private void handleNewEvent(OrderEventDTO event, Ulid currentUlid, Ulid lastUlid) {
        // 检查是否连续事件
        if (currentUlid.compareTo(lastUlid) == 0) {
            processEvent(event);
        }
        // 处理不连续事件
        else {
            handleDiscontinuousEvent(event, currentUlid, lastUlid);
        }
    }

    /**
     * 处理连续事件
     */
    private void processEvent(OrderEventDTO event) {
        try {
            // 幂等检查
            String eventKey = EVENT_KEY_PREFIX + event.getEventId();
            RBucket<String> eventBucket = redissonClient.getBucket(eventKey);

            if (eventBucket.isExists()) {
                log.debug("Skipping duplicate event: {}", event.getEventId());
                return;
            }

            // 保存事件
            eventBucket.set(objectMapper.writeValueAsString(event), 7, TimeUnit.DAYS);

            // 业务处理
            processEventContent(event);

            // 更新最后处理ID
            updateLastProcessedId(event.getEventId());

        } catch (JsonProcessingException e) {
            log.error("JSON serialization error: {}", e.getMessage());
        }
    }


    /**
     * 处理不连续事件
     */
    private void handleDiscontinuousEvent(OrderEventDTO event, Ulid currentUlid, Ulid lastUlid) {
        try {
            // 幂等检查
            String eventKey = EVENT_KEY_PREFIX + event.getEventId();
            RBucket<String> eventBucket = redissonClient.getBucket(eventKey);

            if (!eventBucket.isExists()) {
                eventBucket.set(objectMapper.writeValueAsString(event), 7, TimeUnit.DAYS);
                log.warn("Out-of-order event saved: {}", event.getEventId());

                // 记录缺失范围
                String gapStart = lastUlid.toString();
                String gapEnd = currentUlid.toString();
                compensationService.recordMissingRange(gapStart, gapEnd);
            }

            // 立即触发补偿
            compensationService.compensateMissingEvents();

        } catch (Exception e) {
            log.error("Out-of-order handling failed: {}", event.getEventId(), e);
        }
    }

    /**
     * 获取最后处理的ULID
     */
    private Ulid getLastProcessedUlid() {
        RBucket<String> lastIdBucket = redissonClient.getBucket(LAST_PROCESSED_ID_KEY);
        return Ulid.from(lastIdBucket.get());
    }

    /**
     * 更新最后处理的ID
     */
    public void updateLastProcessedId(String eventId) {
        RBucket<String> lastIdBucket = redissonClient.getBucket(LAST_PROCESSED_ID_KEY);
        lastIdBucket.set(eventId);
    }

    /**
     * 连接断开时调度重连
     */
    private void scheduleReconnect() {
        Executors.newSingleThreadScheduledExecutor().schedule(
                () -> {
                    log.info("Attempting SSE reconnection...");
                    connectSSE();
                },
                30, // 30秒后重试
                TimeUnit.SECONDS
        );
    }

    // SSE事件处理
    @Override
    public void onSSEOpen(EventSource eventSource) {
        currentSSE = eventSource.sse();
        log.info("SSE connection opened");
        redissonClient.getBucket(CONNECTION_STATUS_KEY).set("connected");
    }

    @Override
    public void onSSEClose(EventSource eventSource) {
        log.warn("SSE connection closed");
        currentSSE = null;
        redissonClient.getBucket(CONNECTION_STATUS_KEY).set("disconnected");
        scheduleReconnect();
    }

    @Override
    public void onMessage(EventSource event, String name, String value) {
        try {
            OrderEventDTO eventDTO = objectMapper.readValue(value, OrderEventDTO.class);
            log.debug("Received SSE event: {}", eventDTO.getEventId());
            handleSSEMessage(eventDTO);
        } catch (JsonProcessingException e) {
            log.error("SSE message parsing error: {}", value, e);
        }
    }

    /**
     * 业务处理逻辑（示例）
     */
    private void processEventContent(OrderEventDTO event) {
        // 实际业务处理逻辑
        log.info("Processing event {}: {}", event.getEventId(), event.getEventType());

        // 示例处理：
        // switch (event.getEventType()) {
        //     case "new":
        //         orderService.createOrder(event.getOrder());
        //         break;
        //     case "fill":
        //         orderService.updateOrderFill(event);
        //         break;
        // }
    }
}
